/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.common.io;

import java.io.IOException;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.sss.common.DataReceiver;
import org.sss.common.KeyValues;
import org.sss.mapreduce.Configuration;
import org.sss.mapreduce.DB;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.SssException;
import org.sss.mapreduce.StorageNode;
import org.sss.util.BytesComparator;

public class WorkerFirstKeyValueReader extends AbstractKeyValueReader {
  @SuppressWarnings("unused")
  private static Logger logger = LoggerFactory.getLogger(WorkerFirstKeyValueReader.class);
  private static BytesComparator CMP = new BytesComparator();

  public WorkerFirstKeyValueReader(Configuration conf,
                                   StorageNode node,
                                   GroupID source,
                                   int subhash,
                                   DataReceiver<List<KeyValues>> next) throws SssException {
    super(conf, node, source, subhash, next);
  }

  @Override
  protected void doRead(IOThread iot)
    throws IOException, InterruptedException {
    try {
      DB db = iot.getDB(node);
      int limitNBytes = iot.getIOGetLimitNBytes();
      WorkerQueue q = new WorkerQueue();
      collectWorkers(db, limitNBytes, q);
      workerFirstReadRaw(db, limitNBytes, q);
    } catch (Exception e) {
      // TODO: error handling
      throw new IOException(e);
    }
  }

  static private int nextWorkerIdHead(List<byte[]> bs, byte[] sentinel) {
    int pos   = bs.size() / 2 / 2;
    int begin = 0;
    int end   = bs.size() / 2;
    while (true) {
      int r = CMP.compare(sentinel, bs.get(pos * 2));
      if (r < 0) {
        end = pos;
      }
      else {
        begin = pos + 1;
      }
      if (begin == end) {
        return begin * 2;
      }
      pos = begin + (end - begin) / 2;
    }
  }
  private void collectWorkers(final DB db, final int limitNBytes,
    WorkerQueue workerQueue) throws SssException, IOException {
    final KeyFormatter keyFormatter = KeyFormatter.get(getSource().getKeyFormat());
    assert keyFormatter == KeyFormatter.threadPriorityFormatter: "KeyFormatter is not threadPriorityFormatter.";
    boolean binc = true;
    byte[]       prefix   = KeyFormatter.getPrefix(getSource(), subhash);
    final byte[] sentinel = KeyFormatter.getSentinel(getSource(), subhash);

    while (true) {
      checkCanceled();
      List<byte[]> kvs = isDestructiveRead() ? db.ranged(prefix, sentinel, limitNBytes, binc)
                                             : db.range(prefix, sentinel, limitNBytes, binc);
      countup(kvs);
      binc = false;
      if (kvs.isEmpty()) {
        break;
      }
      byte[] widSentinel = null; // worker id sentinel
      while (kvs.size() > 0) {
        widSentinel = keyFormatter.getWorkerIDSentinel(kvs.get(0));
        int n = nextWorkerIdHead(kvs, widSentinel);
        assert n > 0;
        Worker worker = new Worker(keyFormatter);
        worker.bytes = kvs.subList(0, n);
        worker.sentinel = widSentinel;
        if (n == kvs.size()) {
          worker.next = worker.bytes.get(worker.bytes.size() - 2);
        }
        byte[] key = worker.firstKey();
        workerQueue.push(key, worker);
        kvs = kvs.subList(n, kvs.size());
      }
      prefix = widSentinel;
    }
  }

  private void workerFirstReadRaw(final DB db, final int limitNBytes, WorkerQueue workerQueue)
    throws InstantiationException, IllegalAccessException, Exception {
    final KeyFormatter keyFormatter = KeyFormatter.get(getSource().getKeyFormat());
    assert keyFormatter == KeyFormatter.threadPriorityFormatter: "KeyFormatter is not threadPriorityFormatter.";

    KeyValuesQueue kvq = new KeyValuesQueue(next, limitNBytes);
    try {
      while (!workerQueue.isEmpty()) {
        Map.Entry<byte[], List<Worker>> e = workerQueue.pop();
        byte[] key = e.getKey();
        List<Worker> workers = e.getValue();
        kvq.pushKey(key);
        for (Worker worker: workers) {
          while (true) {
            for ( ; worker.index < worker.bytes.size(); worker.index += 2) {
              if (!keyFormatter.rawKeyEquals(worker.bytes.get(worker.index), key)) {
                break;
              }
              kvq.pushValue(worker.bytes.get(worker.index + 1));

              // for GC
              worker.bytes.set(worker.index, null);
              worker.bytes.set(worker.index + 1, null);
            }
            if (worker.isEmpty() && worker.next != null) {
              checkCanceled();
              List<byte[]> kvs = isDestructiveRead() ? db.ranged(worker.next, worker.sentinel, limitNBytes, false)
                : db.range(worker.next, worker.sentinel, limitNBytes, false);
              int count = kvs.size();
              if (count > 0) {
                countup(kvs);

                worker.next = kvs.get(count - 2);
                worker.bytes = kvs;
                worker.index = 0;
              }
              else {
                worker.next = null;
                worker.bytes = null;
                worker.index = 0;
                break;
              }
            }
            else {
              break;
            }
          }
          if (!worker.isEmpty()) {
            workerQueue.push(worker.firstKey(), worker);
          }
        }
      }
    }
    finally {
      kvq.pushEnd();
    }
  }
}
